package felix.things.router.mqtt.server.broker.internal;

import felix.things.router.mqtt.server.broker.cluster.RedisCluster;
import felix.things.router.mqtt.server.broker.config.BrokerProperties;
import felix.things.router.mqtt.server.broker.service.KafkaService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.lang.management.ManagementFactory;

import static felix.things.router.mqtt.server.store.utils.StoreUtil.getProcessId;

/**
 * 内部通信, 基于发布-订阅范式
 *
 * @author Felix
 * @date 2019/8/21
 */
@Component
@Slf4j
public class InternalCommunication {

    @Autowired
    private BrokerProperties brokerProperties;
    @Autowired
    private KafkaService kafkaService;
    @Autowired
    private RedisCluster redisCluster;

    public void internalSend(InternalMessage internalMessage) {
        String processId = getProcessId("0");
        //broker唯一标识 mqttwk.broker.id
        internalMessage.setBrokerId(brokerProperties.getId());
        internalMessage.setProcessId(processId);
        //如果开启kafka消息转发
        if (brokerProperties.isKafkaBrokerEnabled()) {
            kafkaService.send(internalMessage);
        }
        //如果开启集群功能
        if (brokerProperties.isClusterEnabled()) {
            redisCluster.sendMessage(internalMessage);
        }
    }

}
